package com.caissa.springboot.starter.kafka.consumer;

import com.base.springboot.starter.kafka.consumer.exception.CaissaConsumerDirectStoreErrorMsgException;
import com.base.springboot.starter.kafka.consumer.exception.CaissaConsumerRetryThenStoreErrorMsgException;
import com.base.springboot.starter.kafka.consumer.exception.CaissaIdempotentException;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.transaction.TransactionStatus;
import org.springframework.transaction.support.TransactionCallback;
import org.springframework.transaction.support.TransactionTemplate;

import java.sql.SQLIntegrityConstraintViolationException;

/**
 * 幂等工具类，定义消息重复、业务异常时消息重试、转存的主逻辑
 * User: wangwei
 * Date: 2021-06-15 3:57 下午
 */
@Slf4j
public abstract class CaissaKafkaMsgIdemHelper {

    public <R> R executeWithIdempotentAndRetryAndErrorStore(ConsumerRecord consumerRecord, CBack<R> cBack){
        try {
            return doExecute(consumerRecord, cBack);
        } catch (Exception exception) {
            //把异常扔出去有两种结果：
            //如果是需要重试异常，那么当前消息先将重试3次，重试失败，消息将被写入ES，最终SpringBoot 将自动ack；
            //如果不属于重试异常，消息将直接被写入ES，且自动ack
            //重试异常定义参见 DefaultBackOffRetryTemplate

            if (exception instanceof CaissaIdempotentException) {
                log.info("==========>幂等检查异常，消息处理流程终止，消息ack");
                return null;
            }
            log.info("==========>流程终止，无重试，消息被存储es，消息ack", exception);
            throw new CaissaConsumerDirectStoreErrorMsgException("", exception);
        }
    }

    /**
     *
     * @param consumerRecord
     * @param cBack
     * @param <R>
     * @return
     * @throws Exception
     */
    protected abstract <R> R doExecute(ConsumerRecord consumerRecord, CBack<R> cBack) throws Exception;

    /**
     * 幂等检查
     * @param consumerRecord
     * @throws CaissaIdempotentException 只抛幂等检查异常，自身执行发生异常只记录日志，不影响主业务流程
     */
    protected abstract void checkIdempotent(ConsumerRecord consumerRecord) throws CaissaIdempotentException;

    /**
     * 在子类定义消息业务逻辑
     * @param <R>
     */
    public interface CBack<R>{
        /**
         * 重试逻辑见 DefaultBackOffRetryTemplate
         * 转存逻辑见 ErrMsg2EsHandler
         * @param consumerRecord
         * @param <R> 返回类型
         * @return
         * @throws CaissaIdempotentException 幂等检查异常，消息处理流程终止，消息ack
         * @throws CaissaConsumerDirectStoreErrorMsgException  流程终止，无重试，消息被存储es，消息ack
         * @throws Exception 无重试，直接转存es ，同 CaissaConsumerDirectStoreErrorMsgException
         */
        <R> R doBiz(ConsumerRecord consumerRecord)
                throws CaissaIdempotentException,
                CaissaConsumerDirectStoreErrorMsgException;
    }
}
